package com.itbaizhan.hbase2hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;


/**
 * @Auther: hwh
 * @Date: 2023-02-28 - 02 - 28 - 17:01
 * @Description: com.itbaizhan.hbase2hdfs
 * @version: 1.0
 */
public class HBase2HdfsMain {
    //这是一个main方法，是程序的入口：
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration(true);
        conf.set("mapreduce.framework.name","local");
        conf.set("hbase.zookeeper.quorum","node02,node03,node04");
        Job job = Job.getInstance(conf, "hbase2hdfs demo");
        job.setJarByClass(HBase2HdfsMain.class);
        //从HBase中的sentence表中读取数据
        //可以通过该对象设置查询的列族、列、过滤行等
        Scan scan = new Scan();
        TableMapReduceUtil.initTableMapperJob(
                "sentence",//表名
                scan,
                Hbase2HdfsMapper.class, //指定Mapper类
                Text.class,IntWritable.class, //Mapper类输出的key\value的类型
                job,
                false
        );
        //设置Reducer相关属性
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        job.setReducerClass(Hbase2HdfsReducer.class);
        //设置输出路径
        Path path = new Path("/usr/local/wcout");
        //获取HDFS文件系统的对象
        FileSystem fileSystem = path.getFileSystem(conf);
        //判断输出路径是否存在
        if(fileSystem.exists(path)){
            //如果存在则删除
            fileSystem.delete(path,true);
        }
        FileOutputFormat.setOutputPath(job,path);
        //提交作业
        job.waitForCompletion(true);
    }
}
